/*
 * Copyright (C) 2011 Christopher Probst
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * * Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 *
 * * Redistributions in binary form must reproduce the above copyright
 *   notice, this list of conditions and the following disclaimer in the
 *   documentation and/or other materials provided with the distribution.
 *
 * * Neither the name of the 'FoxNet Codec' nor the names of its 
 *   contributors may be used to endorse or promote products derived
 *   from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
package com.foxnet.codec.impl.bytebuffer;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.WritableByteChannel;

import com.foxnet.codec.Cache;
import com.foxnet.codec.CachePolicy;
import com.foxnet.codec.Codec;
import com.foxnet.codec.CodecRegistry;
import com.foxnet.codec.ObjectOutputChannel;
import com.foxnet.codec.impl2.Flag;
import com.foxnet.codec.impl2.Protocol;

/**
 * @author Christopher Probst
 */
public final class DefaultNIO2Output extends AbstractNIO2BufferedResource
	implements OutputStream {

    // The writable byte channel of this output
    private WritableByteChannel writableByteChannel;
    // The cache class
    private final Class<? extends Cache> cacheClass;
    // The cache of this output
    private Cache lazyCache;
    // Used to clear the cache
    private boolean clearCache;
    // The cache policy of this output!
    private CachePolicy cachePolicy;
    // Used to write serializable objects
    private ObjectOutputStream lazyObjectOutputStream;

    private void writeString(String string) throws IOException {

	// The string length
	int maxChars = string.length();

	// Write char length
	writeInt(maxChars);

	// Write all strings
	for (int i = 0, c; i < maxChars; i++) {
	    // Get char
	    c = string.charAt(i);

	    // Check flush
	    if (limit - pos < 3) {
		flush();
	    }

	    // Write the char as utf
	    if (c <= 0x007F) {
		buffer[pos++] = (byte) c;
	    } else if (c > 0x07FF) {
		buffer[pos] = (byte) (0xE0 | c >> 12 & 0x0F);
		buffer[pos + 1] = (byte) (0x80 | c >> 6 & 0x3F);
		buffer[pos + 2] = (byte) (0x80 | c & 0x3F);
		pos += 3;
	    } else {
		buffer[pos] = (byte) (0xC0 | c >> 6 & 0x1F);
		buffer[pos + 1] = (byte) (0x80 | c & 0x3F);
		pos += 2;
	    }
	}
    }

    private void writeBytes(byte[] bytes) throws IOException {
	// Write array length
	writeInt(bytes.length);

	// Write bytes
	write(bytes);
    }

    private void writeBooleans(boolean[] booleans) throws IOException {
	// Write array length
	writeInt(booleans.length);

	// Write all booleans
	for (boolean b : booleans) {
	    writeBoolean(b);
	}
    }

    private void writeChars(char[] chars) throws IOException {
	// Write array length
	writeInt(chars.length);

	CharBuffer cb = CharBuffer.wrap(chars);
	int oldLimit = cb.limit();
	CharBuffer outCB = buffer.asCharBuffer();
	while (cb.hasRemaining()) {
	    int amount = Math.min(outCB.remaining() / 2, cb.remaining());
	    cb.limit(cb.position() + amount);
	    outCB.put(cb);
	    flush();
	    cb.limit(oldLimit);
	}
    }

    private void writeShorts(short[] shorts) throws IOException {
	// Write array length
	writeInt(shorts.length);

	// Write all shorts
	for (short s : shorts) {
	    writeShort(s);
	}
    }

    private void writeInts(int[] ints) throws IOException {
	// Write array length
	writeInt(ints.length);

	// Write all ints
	for (int i : ints) {
	    writeInt(i);
	}
    }

    private void writeLongs(long[] longs) throws IOException {
	// Write array length
	writeInt(longs.length);

	// Write all longs
	for (long l : longs) {
	    writeLong(l);
	}
    }

    private void writeFloats(float[] floats) throws IOException {
	// Write array length
	writeInt(floats.length);

	// Write all floats
	for (float f : floats) {
	    writeFloat(f);
	}
    }

    private void writeDoubles(double[] doubles) throws IOException {
	// Write array length
	writeInt(doubles.length);

	// Write all doubles
	for (double d : doubles) {
	    writeDouble(d);
	}
    }

    private void writeSerializableObject(Object object, boolean cacheValid)
	    throws IOException {
	// Write flag
	writeTypeClass(Protocol.TC_SERIALIZABLE_OBJECT, cacheValid);

	// Put here into cache!
	assign(object, cacheValid);

	if (lazyObjectOutputStream == null) {
	    lazyObjectOutputStream = new ObjectOutputStream(new OutputStream() {

		@Override
		public void write(int b) throws IOException {
		    writeByte(b);
		}

		/*
		 * (non-Javadoc)
		 * @see java.io.OutputStream#write(byte[], int, int)
		 */
		@Override
		public void write(byte[] b, int off, int len)
			throws IOException {
		    DefaultOutput.this.write(b, off, len);
		}
	    });
	}

	// Write the object
	lazyObjectOutputStream.writeUnshared(object);

	// Flush the stream
	lazyObjectOutputStream.flush();
    }

    @SuppressWarnings("unchecked")
    private <T> void writeOrdinaryObject(T object, Codec<?>[] codecs,
	    boolean cacheValid) throws Exception {
	// Write flag
	writeTypeClass(Protocol.TC_OBJECT, cacheValid);

	// Get the name of the object class
	String name = object.getClass().getName();

	// Write the class name
	writeObject(name);

	// Put here into cache!
	assign(object, cacheValid);

	for (Codec<?> codec : codecs) {
	    // Encode object
	    if (((Codec<T>) codec).encode(this, object)) {
		break;
	    }
	}
    }

    private void writeObject(Object object, boolean cacheValid)
	    throws Exception {

	// Look for cached objects!
	if (lookupAndWrite(object, cacheValid) != -1) {
	    return;
	}

	// Get hierarchy codecs
	Codec<?>[] codecs = getHierarchyCodecs(object.getClass());

	if (codecs.length > 0) {

	    // Write as ordinary object
	    writeOrdinaryObject(object, codecs, cacheValid);

	} else {
	    /*
	     * If there were no codec we have to use the Java serialization if
	     * possible!
	     */
	    if (object instanceof Serializable) {

		// Write serializable object
		writeSerializableObject(object, cacheValid);
	    } else {
		throw new IOException("There are no bound codecs for " + object
			+ ". In addition: The object does not implement "
			+ "java.io.Serializable, either.");
	    }
	}
    }

    private boolean writePrimitive(Object primitiveObject) throws IOException {

	// Save class temporary
	Class<?> primitiveClass = primitiveObject.getClass();

	// Write primitive content
	if (primitiveClass == Byte.class) {
	    writeByte(Protocol.TC_BYTE);
	    writeByte((Byte) primitiveObject);
	} else if (primitiveClass == Boolean.class) {
	    writeByte(Protocol.TC_BOOLEAN);
	    writeBoolean((Boolean) primitiveObject);
	} else if (primitiveClass == Character.class) {
	    writeByte(Protocol.TC_CHAR);
	    writeChar((Character) primitiveObject);
	} else if (primitiveClass == Short.class) {
	    writeByte(Protocol.TC_SHORT);
	    writeShort((Short) primitiveObject);
	} else if (primitiveClass == Integer.class) {
	    writeByte(Protocol.TC_INT);
	    writeInt((Integer) primitiveObject);
	} else if (primitiveClass == Long.class) {
	    writeByte(Protocol.TC_LONG);
	    writeLong((Long) primitiveObject);
	} else if (primitiveClass == Float.class) {
	    writeByte(Protocol.TC_FLOAT);
	    writeFloat((Float) primitiveObject);
	} else if (primitiveClass == Double.class) {
	    writeByte(Protocol.TC_DOUBLE);
	    writeDouble((Double) primitiveObject);
	} else {
	    return false;
	}

	return true;
    }

    private void writeClass(Object classObject, boolean cacheValid)
	    throws Exception {

	// Try the cache
	if (lookupAndWrite(classObject, cacheValid) != -1) {
	    // Return if class already exists
	    return;
	}

	// Write type constant
	writeTypeClass(Protocol.TC_CLASS, cacheValid);

	// Write enum class name
	writeObject(((Class<?>) classObject).getName());

	// Put
	assign(classObject, cacheValid);
    }

    private void writeEnum(Object enumObject, boolean cacheValid)
	    throws Exception {

	// Try the cache
	if (lookupAndWrite(enumObject, cacheValid) != -1) {
	    // Return if enum already exists
	    return;
	}

	// Write type constant
	writeTypeClass(Protocol.TC_ENUM, cacheValid);

	// Write enum class name
	writeObject(enumObject.getClass().getName());

	// Write the ordinal of the enum
	writeInt(((Enum<?>) enumObject).ordinal());

	// Put
	assign(enumObject, cacheValid);
    }

    private void writeArray(Object arrayObject, boolean cacheValid)
	    throws Exception {
	// Try the cache
	if (lookupAndWrite(arrayObject, cacheValid) != -1) {
	    return;
	}

	// Try to get the real array type
	Class<?> arrayType = arrayObject.getClass().getComponentType();

	// 1 means that this is the last array
	int dimsLeft = 1;

	// Go through for all dimensions left
	while (arrayType.isArray()) {
	    // Get next
	    arrayType = arrayType.getComponentType();

	    // Increase
	    dimsLeft++;
	}

	// If 'array' has one dimension and is a primitive type
	if (arrayType.isPrimitive()) {

	    // Put into cache
	    assign(arrayObject, cacheValid);

	    // Write type constant
	    writeTypeClass(Protocol.TC_PRIMITIVE_ARRAY, cacheValid);

	    // Write primitive array content
	    if (arrayType == Byte.TYPE) {
		writeByte(Protocol.TC_BYTE);
		writeBytes((byte[]) arrayObject);
	    } else if (arrayType == Boolean.TYPE) {
		writeByte(Protocol.TC_BOOLEAN);
		writeBooleans((boolean[]) arrayObject);
	    } else if (arrayType == Character.TYPE) {
		writeByte(Protocol.TC_CHAR);
		writeChars((char[]) arrayObject);
	    } else if (arrayType == Short.TYPE) {
		writeByte(Protocol.TC_SHORT);
		writeShorts((short[]) arrayObject);
	    } else if (arrayType == Integer.TYPE) {
		writeByte(Protocol.TC_INT);
		writeInts((int[]) arrayObject);
	    } else if (arrayType == Long.TYPE) {
		writeByte(Protocol.TC_LONG);
		writeLongs((long[]) arrayObject);
	    } else if (arrayType == Float.TYPE) {
		writeByte(Protocol.TC_FLOAT);
		writeFloats((float[]) arrayObject);
	    } else if (arrayType == Double.TYPE) {
		writeByte(Protocol.TC_DOUBLE);
		writeDoubles((double[]) arrayObject);
	    }
	} else {
	    // Write type constant
	    writeTypeClass(Protocol.TC_ARRAY, cacheValid);

	    // Convert to real array for fast access
	    Object[] realArray = (Object[]) arrayObject;

	    // Put dimensions
	    writeByte(dimsLeft);

	    // Write length
	    writeInt(realArray.length);

	    // Write array component type
	    writeObject(arrayType.getName());

	    // Just put into cache!
	    assign(arrayObject, cacheValid);

	    // For all items
	    for (Object item : realArray) {

		// Write next
		writeObject(item);
	    }
	}
    }

    private void writeString(Object stringObject, boolean cacheValid)
	    throws IOException {
	// Convert
	String string = (String) stringObject;

	// Get string length
	int length = string.length();

	// Do not write empty strings!
	if (length == 0) {
	    writeByte(Protocol.TC_EMPTY_STRING);
	    return;
	}

	// Try the cache
	if (lookupAndWrite(string, cacheValid) != -1) {
	    return;
	} else {
	    assign(string, cacheValid);
	}

	// Write string identifier
	writeTypeClass(Protocol.TC_STRING, cacheValid);

	// Write
	writeString(string);
    }

    private WritableByteChannel verifyWritableByteChannel() {
	if (writableByteChannel == null) {
	    throw new IllegalStateException(
		    "The writable byte channel has not " + "be set yet.");
	}
	return writableByteChannel;
    }

    private Cache getLazyCache() {
	if (lazyCache == null) {
	    // Create cache on demand
	    try {
		lazyCache = cacheClass.newInstance();
	    } catch (Exception e) {
		throw new RuntimeException("Failed to create an instance "
			+ "of cache class " + cacheClass + ". Reason: "
			+ e.getMessage(), e);
	    }
	}

	return lazyCache;
    }

    private void writeTypeClass(int tc, boolean cacheObject) throws IOException {
	if (cacheObject) {
	    /*
	     * Very important! If the cache has been cleared we have to inform
	     * the input!
	     */
	    if (clearCache) {
		// Write clear cache command
		writeByte(Protocol.TC_CLEAR_CACHE);

		// Set clear cache to false
		clearCache = false;
	    }

	    // Set the cache flag
	    tc = Flag.set(tc, Protocol.CACHE_FLAG);
	}

	// Write type class!
	writeByte(tc);
    }

    private void writeCacheRef(int index) throws IOException {
	// Write type constant
	writeByte(Protocol.TC_CACHE_REF);

	// Write the index
	writeInt(index);
    }

    private int lookupAndWrite(Object obj, boolean cacheValid)
	    throws IOException {
	if (!cacheValid || lazyCache == null) {
	    return -1;
	}

	// Get index
	int index = lazyCache.lookup(obj);

	// Write cache reference
	if (index != -1) {
	    writeCacheRef(index);
	}

	return index;
    }

    private int assign(Object obj, boolean cacheValid) throws IOException {
	if (!cacheValid) {
	    return -1;
	}

	return getLazyCache().assign(obj);
    }

    public DefaultNIO2Output(CodecRegistry codecSystem, int bufferSize,
	    WritableByteChannel writableByteChannel,
	    Class<? extends CachePolicy> cachePolicyClass) throws IOException {
	super(codecSystem, bufferSize);

	// Save the channel
	this.writableByteChannel = writableByteChannel;

	// Get the cache class
	cacheClass = codecSystem.getDefaultCacheClass();

	// Set cache policy class
	setCachePolicyClass(cachePolicyClass);
    }

    /*
     * (non-Javadoc)
     * @see java.io.Flushable#flush()
     */
    @Override
    public void flush() throws IOException {
	// Verify
	WritableByteChannel tmp = verifyWritableByteChannel();

	// Flip buffer
	buffer.flip();

	while (buffer.hasRemaining()) {
	    // Write bytes
	    tmp.write(buffer);
	}

	buffer.clear();
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeByte(int)
     */
    @Override
    public void writeByte(int byteValue) throws IOException {
	if (buffer.remaining() < 1) {
	    flush();
	}

	buffer.put((byte) byteValue);
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeBoolean(boolean)
     */
    @Override
    public void writeBoolean(boolean booleanValue) throws IOException {
	if (buffer.remaining() < 1) {
	    flush();
	}

	buffer.put((byte) (booleanValue ? 1 : 0));
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeChar(int)
     */
    @Override
    public void writeChar(int charValue) throws IOException {
	if (buffer.remaining() < 2) {
	    flush();
	}

	buffer.putChar((char) charValue);
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeShort(int)
     */
    @Override
    public void writeShort(int shortValue) throws IOException {
	if (buffer.remaining() < 2) {
	    flush();
	}

	buffer.putShort((short) shortValue);
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeInt(int)
     */
    @Override
    public void writeInt(int intValue) throws IOException {
	if (buffer.remaining() < 4) {
	    flush();
	}

	buffer.putInt(intValue);
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeLong(long)
     */
    @Override
    public void writeLong(long longValue) throws IOException {
	if (buffer.remaining() < 8) {
	    flush();
	}

	buffer.putLong(longValue);
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeFloat(float)
     */
    @Override
    public void writeFloat(float floatValue) throws IOException {
	if (buffer.remaining() < 4) {
	    flush();
	}

	buffer.putFloat(floatValue);
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeDouble(double)
     */
    @Override
    public void writeDouble(double doubleValue) throws IOException {
	if (buffer.remaining() < 4) {
	    flush();
	}

	buffer.putDouble(doubleValue);
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#write(byte[])
     */
    @Override
    public void write(byte[] bytes) throws IOException {
	write(bytes, 0, bytes.length);
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#write(byte[], int, int)
     */
    @Override
    public void write(byte[] bytes, int offset, int length) throws IOException {
	flush();
	verifyWritableByteChannel().write(
		ByteBuffer.wrap(bytes, offset, length));
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#getDestination()
     */
    @Override
    public WritableByteChannel getDestination() {
	return writableByteChannel;
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#getDestination(java.lang.Class)
     */
    @SuppressWarnings("unchecked")
    @Override
    public <T extends WritableByteChannel> T getDestination(Class<T> clazz) {
	return (T) writableByteChannel;
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#cacheSize()
     */
    @Override
    public int cacheSize() {
	throw new UnsupportedOperationException();
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#clearCache()
     */
    @Override
    public void clearCache() {
	throw new UnsupportedOperationException();
    }

    /*
     * (non-Javadoc)
     * @see java.io.Closeable#close()
     */
    @Override
    public void close() throws IOException {
	flush();
	verifyWritableByteChannel().close();
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#getCacheClass()
     */
    @Override
    public Class<? extends Cache> getCacheClass() {
	return cacheClass;
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#getCachePolicyClass()
     */
    @Override
    public Class<? extends CachePolicy> getCachePolicyClass() {
	return cachePolicy != null ? cachePolicy.getClass() : null;
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#setCachePolicyClass(java.lang.Class)
     */
    @Override
    public Class<? extends CachePolicy> setCachePolicyClass(
	    Class<? extends CachePolicy> cachePolicyClass) {
	if (cachePolicyClass == null) {
	    throw new NullPointerException("cachePolicyClass cannot be null");
	}

	// Get old cache policy class
	Class<? extends CachePolicy> oldCachePolicyClass =
		getCachePolicyClass();

	// Set new policy
	try {
	    cachePolicy = cachePolicyClass.newInstance();
	} catch (Exception e) {
	    throw new RuntimeException("Failed to create an instance "
		    + "of cache policy class " + cachePolicyClass
		    + ". Reason: " + e.getMessage(), e);
	}

	// Return the old policy
	return oldCachePolicyClass;
    }

    /*
     * (non-Javadoc)
     * @see
     * com.foxnet.codec.Output#setDestination(java.nio.channels.WritableByteChannel
     * )
     */
    @Override
    public WritableByteChannel setDestination(WritableByteChannel destination) {
	WritableByteChannel old = writableByteChannel;
	writableByteChannel = destination;
	return old;
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.BufferedResource#setMinimalBufferSize(int)
     */
    @Override
    public void setMinimalBufferSize(int bufferSize) {
	if (bufferSize < getBufferSize()) {
	    return;
	}

	// Set the buffer
	ByteBuffer newBuffer = ByteBuffer.allocateDirect(bufferSize);
	newBuffer.put((ByteBuffer) buffer.flip());
	buffer = newBuffer;
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeInputStream(java.io.InputStream)
     */
    @Override
    public void writeInputStream(InputStream inputStream) throws IOException {
	throw new UnsupportedOperationException();
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeInputStream(java.io.InputStream, long)
     */
    @Override
    public void writeInputStream(InputStream inputStream, long length)
	    throws IOException {
	throw new UnsupportedOperationException();
    }

    /*
     * (non-Javadoc)
     * @see com.foxnet.codec.Output#writeObject(java.lang.Object)
     */
    @Override
    public void writeObject(Object object) throws Exception {

    }

}
